//! # 消息总线工具
//!
//! 提供消息总线系统的实用工具函数和辅助类

use std::time::{Duration, Instant};
use tokio::time::timeout;

use crate::error::{MessageBusError, MessageBusResult};

/// 重试策略配置
#[derive(Debug, Clone)]
pub struct RetryConfig {
    /// 最大重试次数
    pub max_attempts: u32,
    /// 基础延迟时间
    pub base_delay: Duration,
    /// 最大延迟时间
    pub max_delay: Duration,
    /// 延迟倍数
    pub backoff_multiplier: f64,
    /// 是否启用抖动
    pub jitter: bool,
}

impl Default for RetryConfig {
    fn default() -> Self {
        Self {
            max_attempts: 3,
            base_delay: Duration::from_millis(100),
            max_delay: Duration::from_secs(30),
            backoff_multiplier: 2.0,
            jitter: true,
        }
    }
}

/// 重试器
pub struct Retrier {
    config: RetryConfig,
}

impl Retrier {
    /// 创建新的重试器
    pub fn new(config: RetryConfig) -> Self {
        Self { config }
    }

    /// 执行带重试的操作
    pub async fn retry<T, F, Fut>(&self, operation: F) -> MessageBusResult<T>
    where
        F: Fn() -> Fut,
        Fut: std::future::Future<Output = MessageBusResult<T>>,
    {
        let mut last_error = None;
        
        for attempt in 0..self.config.max_attempts {
            match operation().await {
                Ok(result) => return Ok(result),
                Err(error) => {
                    last_error = Some(error.clone());
                    
                    // 如果是最后一次尝试，直接返回错误
                    if attempt + 1 >= self.config.max_attempts {
                        break;
                    }
                    
                    // 检查错误是否可重试
                    if !error.is_retryable() {
                        return Err(error);
                    }
                    
                    // 计算延迟时间
                    let delay = self.calculate_delay(attempt);
                    tokio::time::sleep(delay).await;
                }
            }
        }
        
        if let Some(error) = last_error {
            Err(MessageBusError::retry_exhausted(
                "operation",
                self.config.max_attempts,
            ))
        } else {
            Err(MessageBusError::internal_error("重试失败"))
        }
    }

    /// 计算延迟时间
    fn calculate_delay(&self, attempt: u32) -> Duration {
        let base_delay_ms = self.config.base_delay.as_millis() as f64;
        let multiplier = self.config.backoff_multiplier.powi(attempt as i32);
        let delay_ms = base_delay_ms * multiplier;
        
        // 应用最大延迟限制
        let delay_ms = delay_ms.min(self.config.max_delay.as_millis() as f64);
        
        // 应用抖动
        let final_delay_ms = if self.config.jitter {
            let jitter_factor = fastrand::f64() * 0.1; // 10% 抖动
            delay_ms * (1.0 + jitter_factor)
        } else {
            delay_ms
        };
        
        Duration::from_millis(final_delay_ms as u64)
    }
}

/// 超时包装器
pub struct TimeoutWrapper {
    timeout_duration: Duration,
}

impl TimeoutWrapper {
    /// 创建新的超时包装器
    pub fn new(timeout_duration: Duration) -> Self {
        Self { timeout_duration }
    }

    /// 执行带超时的操作
    pub async fn execute<T, F, Fut>(&self, operation: F) -> MessageBusResult<T>
    where
        F: FnOnce() -> Fut,
        Fut: std::future::Future<Output = MessageBusResult<T>>,
    {
        match timeout(self.timeout_duration, operation()).await {
            Ok(result) => result,
            Err(_) => Err(MessageBusError::timeout_error(
                "operation",
                self.timeout_duration.as_millis() as u64,
            )),
        }
    }
}

/// 性能监控器
pub struct PerformanceMonitor {
    start_time: Instant,
    operation_name: String,
}

impl PerformanceMonitor {
    /// 开始监控操作
    pub fn start(operation_name: impl Into<String>) -> Self {
        Self {
            start_time: Instant::now(),
            operation_name: operation_name.into(),
        }
    }

    /// 结束监控并返回执行时间
    pub fn finish(self) -> Duration {
        let duration = self.start_time.elapsed();
        tracing::debug!(
            "操作 '{}' 执行时间: {:?}",
            self.operation_name,
            duration
        );
        duration
    }

    /// 结束监控并记录到指标系统
    #[cfg(feature = "metrics")]
    pub async fn finish_with_metrics(
        self,
        metrics: &crate::metrics::MessageBusMetrics,
        metric_name: &str,
    ) -> Duration {
        let duration = self.start_time.elapsed();
        let duration_ms = duration.as_millis() as f64;
        
        let _ = metrics.observe_histogram(metric_name, duration_ms).await;
        
        tracing::debug!(
            "操作 '{}' 执行时间: {:?}",
            self.operation_name,
            duration
        );
        
        duration
    }
}

/// 批处理器 - 用于批量处理消息
pub struct BatchProcessor<T> {
    items: Vec<T>,
    batch_size: usize,
    flush_interval: Duration,
    last_flush: Instant,
}

impl<T> BatchProcessor<T> {
    /// 创建新的批处理器
    pub fn new(batch_size: usize, flush_interval: Duration) -> Self {
        Self {
            items: Vec::with_capacity(batch_size),
            batch_size,
            flush_interval,
            last_flush: Instant::now(),
        }
    }

    /// 添加项目到批次
    pub fn add(&mut self, item: T) -> Option<Vec<T>> {
        self.items.push(item);
        
        if self.items.len() >= self.batch_size {
            Some(self.flush())
        } else {
            None
        }
    }

    /// 检查是否需要刷新
    pub fn should_flush(&self) -> bool {
        !self.items.is_empty() && self.last_flush.elapsed() >= self.flush_interval
    }

    /// 强制刷新批次
    pub fn flush(&mut self) -> Vec<T> {
        let items = std::mem::take(&mut self.items);
        self.last_flush = Instant::now();
        items
    }

    /// 获取当前批次大小
    pub fn len(&self) -> usize {
        self.items.len()
    }

    /// 检查批次是否为空
    pub fn is_empty(&self) -> bool {
        self.items.is_empty()
    }
}

/// 主题匹配工具
pub struct TopicMatcher;

impl TopicMatcher {
    /// 检查主题是否匹配模式
    pub fn matches(pattern: &str, topic: &str) -> bool {
        if pattern == "**" || pattern == "#" {
            return true; // 匹配所有
        }

        if pattern == topic {
            return true; // 精确匹配
        }

        // 通配符匹配
        Self::wildcard_match(pattern, topic)
    }

    /// 通配符匹配实现
    fn wildcard_match(pattern: &str, topic: &str) -> bool {
        // 检测分隔符类型
        let separator = if pattern.contains('/') || topic.contains('/') {
            '/'
        } else {
            '.'
        };
        
        let pattern_parts: Vec<&str> = pattern.split(separator).collect();
        let topic_parts: Vec<&str> = topic.split(separator).collect();

        Self::match_parts(&pattern_parts, &topic_parts, 0, 0)
    }

    /// 递归匹配部分
    fn match_parts(
        pattern_parts: &[&str],
        topic_parts: &[&str],
        pattern_idx: usize,
        topic_idx: usize,
    ) -> bool {
        // 如果模式和主题都到达末尾，匹配成功
        if pattern_idx >= pattern_parts.len() && topic_idx >= topic_parts.len() {
            return true;
        }

        // 如果模式到达末尾但主题还有剩余，匹配失败
        if pattern_idx >= pattern_parts.len() {
            return false;
        }

        let pattern_part = pattern_parts[pattern_idx];

        if pattern_part == "**" || pattern_part == "#" {
            // ** 或 # 匹配零个或多个级别
            if pattern_idx + 1 < pattern_parts.len() {
                // 尝试从当前主题位置开始匹配剩余模式
                for i in topic_idx..=topic_parts.len() {
                    if Self::match_parts(pattern_parts, topic_parts, pattern_idx + 1, i) {
                        return true;
                    }
                }
                false
            } else {
                // ** 或 # 是最后一个模式部分，匹配剩余所有主题
                true
            }
        } else if pattern_part == "*" || pattern_part == "+" {
            // * 或 + 匹配单个级别
            if topic_idx >= topic_parts.len() {
                return false;
            }
            Self::match_parts(pattern_parts, topic_parts, pattern_idx + 1, topic_idx + 1)
        } else {
            // 精确匹配
            if topic_idx >= topic_parts.len() || pattern_part != topic_parts[topic_idx] {
                return false;
            }
            Self::match_parts(pattern_parts, topic_parts, pattern_idx + 1, topic_idx + 1)
        }
    }

    /// 提取主题的级别
    pub fn extract_levels(topic: &str) -> Vec<&str> {
        topic.split('.').collect()
    }

    /// 规范化主题名称
    pub fn normalize_topic(topic: &str) -> String {
        // 移除开头和结尾的点号
        let trimmed = topic.trim_matches('.');
        
        // 合并连续的点号
        let mut result = String::new();
        let mut prev_char = '\0';
        
        for ch in trimmed.chars() {
            if ch == '.' && prev_char == '.' {
                continue; // 跳过连续的点号
            }
            result.push(ch);
            prev_char = ch;
        }
        
        result
    }
}

/// 消息验证器
pub struct MessageValidator;

impl MessageValidator {
    /// 验证主题名称
    pub fn validate_topic(topic: &str) -> MessageBusResult<()> {
        if topic.is_empty() {
            return Err(MessageBusError::message_format_error("主题不能为空"));
        }

        if topic.len() > 256 {
            return Err(MessageBusError::message_format_error("主题长度不能超过256字符"));
        }

        // 检查非法字符
        for ch in topic.chars() {
            if ch.is_control() || ch == '\0' {
                return Err(MessageBusError::message_format_error(
                    format!("主题包含非法字符: {:?}", ch)
                ));
            }
        }

        Ok(())
    }

    /// 验证订阅模式
    pub fn validate_pattern(pattern: &str) -> MessageBusResult<()> {
        if pattern.is_empty() {
            return Err(MessageBusError::message_format_error("订阅模式不能为空"));
        }

        if pattern.len() > 256 {
            return Err(MessageBusError::message_format_error("订阅模式长度不能超过256字符"));
        }

        // 检查通配符使用是否正确
        let parts: Vec<&str> = pattern.split('.').collect();
        for (i, part) in parts.iter().enumerate() {
            if *part == "**" || *part == "#" {
                // ** 或 # 只能出现在最后
                if i != parts.len() - 1 {
                    return Err(MessageBusError::message_format_error(
                        "多级通配符只能出现在模式末尾"
                    ));
                }
            }
        }

        Ok(())
    }

    /// 验证消息大小
    pub fn validate_message_size(size: usize, max_size: usize) -> MessageBusResult<()> {
        if size > max_size {
            return Err(MessageBusError::message_format_error(
                format!("消息大小 {} 超过最大限制 {}", size, max_size)
            ));
        }
        Ok(())
    }
}

#[cfg(test)]
mod tests {
    use super::*;

    #[tokio::test]
    async fn test_retrier() {
        let config = RetryConfig {
            max_attempts: 3,
            base_delay: Duration::from_millis(10),
            ..Default::default()
        };
        let retrier = Retrier::new(config);

        let attempt_count = std::sync::Arc::new(std::sync::atomic::AtomicU32::new(0));
        let count_clone = attempt_count.clone();
        
        let result = retrier.retry(move || {
            let count = count_clone.clone();
            async move {
                let current = count.fetch_add(1, std::sync::atomic::Ordering::Relaxed) + 1;
                if current < 3 {
                    Err(MessageBusError::network_error("临时失败"))
                } else {
                    Ok("成功")
                }
            }
        }).await;

        assert!(result.is_ok());
        assert_eq!(result.unwrap(), "成功");
        assert_eq!(attempt_count.load(std::sync::atomic::Ordering::Relaxed), 3);
    }

    #[tokio::test]
    async fn test_timeout_wrapper() {
        let timeout_wrapper = TimeoutWrapper::new(Duration::from_millis(100));

        // 正常操作
        let result = timeout_wrapper.execute(|| async {
            tokio::time::sleep(Duration::from_millis(50)).await;
            Ok::<&str, MessageBusError>("成功")
        }).await;
        assert!(result.is_ok());

        // 超时操作
        let result = timeout_wrapper.execute(|| async {
            tokio::time::sleep(Duration::from_millis(200)).await;
            Ok::<&str, MessageBusError>("成功")
        }).await;
        assert!(result.is_err());
    }

    #[test]
    fn test_batch_processor() {
        let mut processor = BatchProcessor::new(3, Duration::from_secs(1));

        assert!(processor.add("item1").is_none());
        assert!(processor.add("item2").is_none());
        
        let batch = processor.add("item3");
        assert!(batch.is_some());
        assert_eq!(batch.unwrap(), vec!["item1", "item2", "item3"]);
        assert!(processor.is_empty());
    }

    #[test]
    fn test_topic_matcher() {
        // 精确匹配
        assert!(TopicMatcher::matches("user.events.login", "user.events.login"));
        assert!(!TopicMatcher::matches("user.events.login", "user.events.logout"));

        // 单级通配符
        assert!(TopicMatcher::matches("user.events.*", "user.events.login"));
        assert!(TopicMatcher::matches("user.events.*", "user.events.logout"));
        assert!(!TopicMatcher::matches("user.events.*", "user.profile.update"));

        // 多级通配符
        assert!(TopicMatcher::matches("user.**", "user.events.login"));
        assert!(TopicMatcher::matches("user.**", "user.profile.update"));
        assert!(TopicMatcher::matches("**", "any.topic.here"));

        // MQTT 风格通配符
        assert!(TopicMatcher::matches("user/+/events", "user/123/events"));
        assert!(TopicMatcher::matches("user/#", "user/123/events/login"));
    }

    #[test]
    fn test_topic_normalization() {
        assert_eq!(TopicMatcher::normalize_topic(".user..events."), "user.events");
        assert_eq!(TopicMatcher::normalize_topic("user...events"), "user.events");
        assert_eq!(TopicMatcher::normalize_topic("normal.topic"), "normal.topic");
    }

    #[test]
    fn test_message_validator() {
        // 有效主题
        assert!(MessageValidator::validate_topic("user.events.login").is_ok());
        assert!(MessageValidator::validate_topic("a").is_ok());

        // 无效主题
        assert!(MessageValidator::validate_topic("").is_err());
        assert!(MessageValidator::validate_topic(&"x".repeat(300)).is_err());

        // 有效模式
        assert!(MessageValidator::validate_pattern("user.events.*").is_ok());
        assert!(MessageValidator::validate_pattern("user.**").is_ok());

        // 无效模式
        assert!(MessageValidator::validate_pattern("").is_err());
        assert!(MessageValidator::validate_pattern("user.**.events").is_err());
    }

    #[test]
    fn test_performance_monitor() {
        let monitor = PerformanceMonitor::start("test_operation");
        std::thread::sleep(Duration::from_millis(10));
        let duration = monitor.finish();
        assert!(duration >= Duration::from_millis(10));
    }
}